package org.budo.dubbo.registry.redis;

import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

import org.budo.dubbo.registry.redis.util.DubboUrlUtil;
import org.budo.dubbo.registry.redis.util.RegistryJedisUtil;

import com.alibaba.dubbo.common.Constants;
import com.alibaba.dubbo.common.URL;
import com.alibaba.dubbo.common.logger.Logger;
import com.alibaba.dubbo.common.logger.LoggerFactory;
import com.alibaba.dubbo.common.utils.CollectionUtils;
import com.alibaba.dubbo.common.utils.StringUtils;
import com.alibaba.dubbo.registry.NotifyListener;

/**
 * @author li
 * @version 1 (2015年11月14日 上午11:10:34)
 * @since Java7
 * @see com.alibaba.dubbo.registry.redis.RedisRegistry
 * @see com.alibaba.dubbo.config.ReferenceConfig#createProxy()
 * @see com.alibaba.dubbo.config.AbstractInterfaceConfig#loadRegistries(boolean)
 */
public class BudoRedisRegistry extends AbstractBudoRedisRegistry {
    private static final Logger log = LoggerFactory.getLogger(BudoRedisRegistry.class);

    /**
     * 每一个Url会初始化一个JedisRegistry对象,应该依赖于Factory 每个Service初始化一个对象 register 一次 dubbo:// subscribe 一次 provider:// listener 每次均不同 引用端 每个引用初始化一个 register 一次 consumer://
     * 
     * @see com.alibaba.dubbo.registry.redis.RedisRegistry#RedisRegistry(URL)
     */
    public BudoRedisRegistry(URL url) {
        super(url);

        int expirePeriod = url.getParameter(Constants.SESSION_TIMEOUT_KEY, Constants.DEFAULT_SESSION_TIMEOUT);

        ScheduledFuture<?> expireFuture = expireExecutor.scheduleWithFixedDelay(new Runnable() {
            public void run() {
                BudoRedisRegistry.this.deferExpired(); // 延长过期时间
            }
        }, expirePeriod / 2, expirePeriod / 2, TimeUnit.MILLISECONDS);

        this.expireFuture = expireFuture;
    }

    /**
     * 延长有效期
     */
    private void deferExpired() {
        try {
            Set<URL> registered = this.getRegistered();
            Set<URL> registeredUrls = new HashSet<URL>(registered);
            for (URL registeredUrl : registeredUrls) {
                boolean dynamicKey = registeredUrl.getParameter(Constants.DYNAMIC_KEY, true);
                if (dynamicKey) {
                    long reply = redisRegistryRepository.register(this, registeredUrl); // 重复注册, 延长服务时间

                    if ((int) reply == 1) { // 是新增
                        redisRegistryRepository.publishRegister(this, registeredUrl); // 发通知
                    }

                    redisRegistryRepository.cleanExpiredKeys(this, registeredUrl); // 清除过期数据
                }
            }
        } catch (Throwable e) {
            log.error("#72 deferExpired error, e=" + e, e);
        }
    }

    /**
     * 注册中心是否可达, 客户端会访问
     * 
     * @see com.alibaba.dubbo.registry.redis.RedisRegistry#isAvailable()
     */
    public boolean isAvailable() {
        return redisRegistryRepository.isAvailable(this);
    }

    /**
     * 销毁
     * 
     * @see com.alibaba.dubbo.registry.redis.RedisRegistry#destroy()
     */
    public void destroy() {
        log.info("destroy, this=" + this.getUrl().toServiceString() + ", destroy");
        super.destroy();

        // this.destroyJedisPool(); // 连接池是静态的共享的, 不能被销毁掉
        this.cancelExpireFuture(true);
    }

    /**
     * 不调用 failback 的, 将之复制一份再修改
     * 
     * @see com.alibaba.dubbo.registry.support.FailbackRegistry#register(URL)
     */
    public void register(URL url) {
        this.failbackRegistry_register(url);
    }

    /**
     * 注册, 客户端会访问
     * 
     * @see com.alibaba.dubbo.registry.redis.RedisRegistry#doRegister(URL)
     */
    public void doRegister(URL url) {
        log.info("#105 doRegister, jedis.active=" + this.getJedisPool().getNumActive() + ", jedis.waiters=" + this.getJedisPool().getNumWaiters() //
                + ", this=" + this + ", url=" + url.toServiceString());

        redisRegistryRepository.register(this, url);
        redisRegistryRepository.publishRegister(this, url);
    }

    /**
     * 取消注册
     * 
     * @see com.alibaba.dubbo.registry.redis.RedisRegistry#doUnregister(URL)
     */
    public void doUnregister(URL url) {
        log.info("doUnregister, this=" + this.getUrl().toServiceString() + ", url=" + url.toServiceString());

        redisRegistryRepository.unregister(this, url);
        redisRegistryRepository.publishUnregister(this, url);
    }

    /**
     * 不调用 failback 的, 将之复制一份再修改
     * 
     * @see com.alibaba.dubbo.registry.support.FailbackRegistry#subscribe(URL, NotifyListener)
     */
    public void subscribe(URL url, NotifyListener listener) {
        this.failbackRegistry_subscribe(url, listener);
    }

    /**
     * 订阅
     * 
     * @see com.alibaba.dubbo.registry.redis.RedisRegistry#doSubscribe(URL, NotifyListener)
     */
    public void doSubscribe(URL url, NotifyListener notifyListener) {
        log.info("#147 doSubscribe, this=" + this + ", subscribe, url=" + url.toServiceString() + ", notifyListener=" + notifyListener + ", version=" + url.getParameter("version"));

        BudoRedisRegistryPubSub jedisRegistryPubSub = this.getJedisRegistryPubSub();
        Map<URL, NotifyListener> notifyListenerMap = jedisRegistryPubSub.getNotifyListenerMap();
        notifyListenerMap.put(url, notifyListener); // 从现在开始监听

        // TODO 是否可以放入线程池异步
        // consumer 端后启动, 未收到服务端的通知, 这里补通知一遍, 难道不是应该由心跳来完成这个通信 ?
        if (StringUtils.isEquals(Constants.CONSUMER_PROTOCOL, url.getProtocol())) {
            List<URL> notifyUrls = redisRegistryRepository.lookup(this, url);
            if (!CollectionUtils.isEmpty(notifyUrls)) {
                log.info("#157 doSubscribe, jedis.active=" + this.getJedisPool().getNumActive() + ", jedis.waiters=" + this.getJedisPool().getNumWaiters() //
                        + ", looking for providers, url=" + DubboUrlUtil.toSimpleString(url) + ", notifyUrls=" + DubboUrlUtil.toSimpleString(notifyUrls));
                notifyListener.notify(notifyUrls);
            }
        }
    }

    /**
     * 取消订阅
     * 
     * @see com.alibaba.dubbo.registry.redis.RedisRegistry#doUnsubscribe(URL, NotifyListener)
     */
    public void doUnsubscribe(URL url, NotifyListener listener) {
        log.info("doUnsubscribe, this=" + this.getUrl().toServiceString() + ", listener=" + listener + ", unsubscribe, url=" + url.toServiceString());
    }

    public List<URL> lookup(URL url) {
        List<URL> lookup = new ArrayList<URL>();

        List<URL> redisLookup = redisRegistryRepository.lookup(this, url);
        List<URL> superLookup = super.lookup(url);

        if (null != redisLookup) {
            lookup.addAll(redisLookup);
        }

        if (null != superLookup) {
            lookup.addAll(superLookup);
        }

        return lookup;
    }

    public String toString() {
        return this.getClass().getName() + "@" + Integer.toHexString(this.hashCode()) + ", " //
                + this.getProtocol() + "://" + "redis-no-user" + "@" + RegistryJedisUtil.getHost(this) + ":" + RegistryJedisUtil.getPort(this);
    }
}